/**
 * 有问题
 *
 *
 * 列表状态
 *
 * 目录那块好像有问题
 */
package com.atguigu.day5



import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.api.common.state.{ListStateDescriptor, ValueStateDescriptor}
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer



object ListStateExceple {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.enableCheckpointing(1000L)
    env.setStateBackend(new FsStateBackend("file:///D://job//idea//idea2018_workspces//flink//src//main//resources//checkpoint"))

    val value = env.addSource(new SensorSource)
      .filter(_.id.equals("sensor_1"))
      .keyBy(_.id)
      .process(new Keyed)
      value.print()
   env.execute()
}
  class Keyed extends KeyedProcessFunction[String,SensorReading,String]{
    lazy val listState = getRuntimeContext.getListState(
      new ListStateDescriptor[SensorReading]("list-state",Types.of[SensorReading])
    )
    lazy val timer = getRuntimeContext.getState(
      new ValueStateDescriptor[Long]("timer",Types.of[Long])
    )

    override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
      listState.add(value)//
      if(timer.value() == 0L){
        val ts = ctx.timerService().currentProcessingTime()+10*1000L
        ctx.timerService().registerProcessingTimeTimer(ts)
        timer.update(ts)
      }
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
      //不能直接对列表状态变量进行计数
      val readings:ListBuffer[SensorReading]=new ListBuffer[SensorReading]
      //隐式类型转换必须导入
      import scala.collection.JavaConversions._
      for (r <- listState.get()){
        readings += r
      }
      out.collect("当前时刻列表状态变量里面共有"+readings.size+"条数据")
      timer.clear()
    }
  }
}